package assignmentImplementation;

import java.io.BufferedReader;
import java.io.DataInputStream;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.StringTokenizer;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import Locker.Locker;

import keyValueBaseExceptions.BeginGreaterThanEndException;
import keyValueBaseExceptions.KeyAlreadyPresentException;
import keyValueBaseExceptions.KeyNotFoundException;
import keyValueBaseInterfaces.Index;
import keyValueBaseInterfaces.Pair;


public class IndexImpl implements Index<KeyImpl,ValueListImpl>
{
	
	private final ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
	//Possibly we need to use the locks in the index, instead.
	private final Lock write = readWriteLock.writeLock();

	private Locker locker;
	private StoreImpl store;
	private ValueSerializerImpl valSer;
	private int counter;
	/*
	 * in out pair, the long refers to the initial memory address that our data
	 * has been saved too, and the integer refers to the length of the data in the file 
 	 */
	
	private HashMap<KeyImpl,Pair<Long,Integer>> m;
	private long endAddr;
	

	public IndexImpl(){

		this.counter = 0;
		valSer = new ValueSerializerImpl();
		endAddr = 0;
		m = new HashMap<KeyImpl,Pair<Long,Integer>>();
		locker = new Locker();
	}
	
	public void insert(KeyImpl k, ValueListImpl v)
			throws KeyAlreadyPresentException, IOException {
		locker.WriteLock(k);
		try{
			if(m.containsKey(k)){
				throw new KeyAlreadyPresentException(k);
			}
			else{
				long tempEndAddr;
				byte[] temp = valSer.toByteArray(v);
				//we are using the ReentrantReadWriteLock implementation found in java
				write.lock();
				try{
					tempEndAddr = endAddr;
					
					endAddr += temp.length;
				}
				finally{
					write.unlock();
				}
				
				store.write(tempEndAddr, temp);
				m.put(k, new Pair<Long, Integer>(tempEndAddr,temp.length));

			}
		}
		finally
		{
			locker.WriteUnlock(k);
		}
		counter++;
	}

	public void remove(KeyImpl k) throws KeyNotFoundException {
		locker.WriteLock(k);
		try{
			if(!m.containsKey(k)){
				throw new KeyNotFoundException(k);
			}
			else{
				m.remove(k);
			}
		}
		finally {
		locker.WriteUnlock(k);
		}
		counter++;
	}

	public ValueListImpl get(KeyImpl k) throws KeyNotFoundException,
			IOException {
		ValueListImpl aList = null;
		locker.ReadLock(k);	
		try{
			if(!m.containsKey(k)){
				throw new KeyNotFoundException(k);
			}
			else{
			//the first element of the pair is the address, the second one is the length
				aList = valSer.fromByteArray(store.read(m.get(k).getKey(), m.get(k).getValue()));
			}
		}

		finally{
			locker.ReadUnlock(k);
		}
		this.counter++;
		return aList;

	}

	public void update(KeyImpl k, ValueListImpl v) throws KeyNotFoundException,
			IOException {
		locker.WriteLock(k);
		try{
			if(!m.containsKey(k)){
				throw new KeyNotFoundException(k);
			}
			else{
				
				long tempEndAddr;
				byte[] temp = valSer.toByteArray(v);
				//we are using the ReentrantReadWriteLock implementation found in java
				write.lock();
				try{
					tempEndAddr = endAddr;
					
					endAddr += temp.length;
				}
				finally{
					write.unlock();
				}
				
				store.write(tempEndAddr, temp);
				m.put(k, new Pair<Long, Integer>(tempEndAddr,temp.length));
			
							
				}
		}
		finally{
			locker.WriteUnlock(k);
		}
		counter++;
	}

	public List<ValueListImpl> scan(KeyImpl begin, KeyImpl end)
			throws BeginGreaterThanEndException, IOException {
		
		
		if(begin.compareTo(end) == 1){
			throw new BeginGreaterThanEndException(begin, end);
		}
		else{
			ValueListImpl aux = null;
			List<ValueListImpl> list = new ArrayList<ValueListImpl>();
			for(int i=begin.getKey(); i<=end.getKey(); i++){
				KeyImpl k = new KeyImpl(i);
				if(m.containsKey(k)){
					//We do the IndexImpl of get to return the value associated to the key.
					//We do this in order to have an atomic version of the value we are getting
					//for each key.
					try {
						aux = this.get(k);
					} catch (KeyNotFoundException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					list.add( aux );
				}
							//valSer.fromByteArray(
								//	store.read(m.get(k).getKey(), m.get(k).getValue())));
			}
			counter++;
			return list;
		}
	}

	public List<ValueListImpl> atomicScan(KeyImpl begin, KeyImpl end)
			throws BeginGreaterThanEndException, IOException {
		// TODO Auto-generated method stub
		
		if(begin.compareTo(end) == 1){
			throw new BeginGreaterThanEndException(begin, end);
		}
		else
		{
			List<ValueListImpl> list = new ArrayList<ValueListImpl>();
			int b = begin.getKey();
			int e = end.getKey();
			for(int i=b; i<= e;i++)
			{
				KeyImpl k2 = new KeyImpl(i);
				locker.ReadLock(k2);
			}
			try{
				for(int i=b; i<=e; i++){
					KeyImpl k = new KeyImpl(i);
					if(m.containsKey(k))		
						list.add( 
								valSer.fromByteArray(
										store.read(m.get(k).getKey(), m.get(k).getValue())));
				}
			}
			finally{
				for(int i=b; i<= e;i++)
				{
					KeyImpl k2 = new KeyImpl(i);
					locker.ReadUnlock(k2);
				}
			}
			counter++;
			return list;
		}
		
	}

	public void bulkPut(List<Pair<KeyImpl, ValueListImpl>> keys)
			throws IOException {
		//ArrayList<Pair<KeyImpl, ValueListImpl>> orderedKeys;
		Collections.sort(keys);
		
		Iterator<Pair<KeyImpl, ValueListImpl>> iter = keys.iterator();
		while(iter.hasNext())
		{
			Pair<KeyImpl, ValueListImpl> element = iter.next();
			KeyImpl k = element.getKey();
			locker.WriteLock(k);
		}
		try{
			for(int i=0;i<keys.size(); i++){
				
				Pair<KeyImpl, ValueListImpl> element = keys.get(i);
				KeyImpl k = element.getKey();
				if(this.m.containsKey(k))
				{
					//UPDATE CASE
					//this.update(k, element.getValue());
					long tempEndAddr;
					byte[] temp = valSer.toByteArray(element.getValue());
					//we are using the ReentrantReadWriteLock implementation found in java
					write.lock();
					try{
						tempEndAddr = endAddr;
						
						endAddr += temp.length;
					}
					finally{
						write.unlock();
					}
					
					store.write(tempEndAddr, temp);
					m.put(k, new Pair<Long, Integer>(tempEndAddr,temp.length));
				}
				else
				{
					//INSERT CASE
					//this.insert(k, element.getValue());
					long tempEndAddr;
					byte[] temp = valSer.toByteArray(element.getValue());
					//we are using the ReentrantReadWriteLock implementation found in java
					write.lock();
					try{
						tempEndAddr = endAddr;
						
						endAddr += temp.length;
					}
					finally{
						write.unlock();
					}
					
					store.write(tempEndAddr, temp);
					m.put(k, new Pair<Long, Integer>(tempEndAddr,temp.length));
				}
				
			}
		}
		finally{
			iter = keys.iterator();
			
			while(iter.hasNext())
			{
				Pair<KeyImpl, ValueListImpl> element = iter.next();
				KeyImpl k = element.getKey();
				locker.WriteUnlock(k);
			}
		}
		counter++;
		
	}
	
	public boolean storeIsInit(){
		if(store==null)
			return false;
		else{
			return store.isInitialized();
		}
		
	}
	
	
	public void StoreInit(String fileName) throws FileNotFoundException {
		
			store = new StoreImpl(fileName);

			
	}

	public synchronized void initFile(String fileName) {
		
		
		try {
			FileInputStream fstream = new FileInputStream(fileName);
			DataInputStream in = new DataInputStream(fstream);
			BufferedReader br = new BufferedReader(new InputStreamReader(in));
			
			String strLine;

			int lineCounter = 1;
			while ((strLine = br.readLine()) != null)   {
				StringTokenizer token = new StringTokenizer(strLine);
				String str;
				if(token.hasMoreTokens()){
					str = token.nextToken();
					if(str.matches("[0-9]+")){
						
						KeyImpl key = new KeyImpl(Integer.parseInt(str));
						if (token.hasMoreTokens()){
							str = token.nextToken();
							ValueImpl value = new ValueImpl();
							value.setValue(str);
							if(token.hasMoreElements()){
								System.out.println("Line " + lineCounter + " has more than 2 values");
							}
							else{
								ValueListImpl vli = new ValueListImpl();
								vli.add(value);
								if(!m.containsKey(key))
									this.insert(key,vli);
								else
									this.addData(key, vli);
							}
						}
						else{
							System.out.println("Key on line " + lineCounter + " has no value");
						}
					}
					else{
						System.out.println("Data on line " + lineCounter + " does not start with an int");
					}
				}
				lineCounter++;
			}
			br.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (KeyAlreadyPresentException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
	
	public void throughputCalc() throws InterruptedException {
		// TODO Auto-generated method stub
		
		while(true){		
			
			Thread.sleep(10000);


			System.out.println("Throughput: "+ ((float) this.counter)/10 +" server operations/sec");
			this.counter = 0;
		}
	}
	
	private void addData(KeyImpl k, ValueListImpl v) {
		
		byte[] updates;
		try {
			updates = valSer.toByteArray(v);

			if(m.get(k).getKey()+m.get(k).getValue() == endAddr) { //if these are the last values in memory
				store.write(endAddr, updates);
				m.put(k, new Pair<Long, Integer>(m.get(k).getKey(),m.get(k).getValue() + updates.length));
				endAddr += updates.length;
			}	
			else{
				byte[] source = store.read(m.get(k).getKey(), m.get(k).getValue());
				byte[] tar = new byte[source.length + updates.length];
				System.arraycopy(source, 0, tar, 0, source.length);
				System.arraycopy(updates, 0, tar, source.length, updates.length);
				store.write(endAddr, tar);
				m.put(k, new Pair<Long, Integer>(this.endAddr,tar.length));
				endAddr += tar.length;
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}